package aeemqx

import (
	"context"
	"devops-super/internal/model/mymodel"
	"devops-super/internal/service"
	"devops-super/utility/util/appUtil"
	"errors"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/gogf/gf/v2/os/glog"
	"log"
	"os"
	"strconv"
	"time"
)

type AeEmqxObj struct {
}

var (
	aeMqttClients                     = make(map[string]mqtt.Client) // 记录所有连接mqtt的客户端 键为host
	f1            mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
		// 异常全局捕获
		defer func() {
			if err := recover(); err != nil {
				globalErrorHandler(fmt.Errorf("%v", err))
			}
		}()
		ctx := context.Background()
		service.Aeemqx().HandleAeMqttData(ctx, msg)
	}
	singleAeEmqxObj *AeEmqxObj
	sharePrefix     = "$share/g/"
	mqttConfigKey   = "mqtt.default"
)

func globalErrorHandler(err error) {
	glog.Line().Info(context.Background(), err.Error())
}

func GetAeEmqxInstance() *AeEmqxObj {
	if singleAeEmqxObj == nil {
		singleAeEmqxObj = new(AeEmqxObj)
	}
	return singleAeEmqxObj
}

func getSubscribers(ctx context.Context, mqttConf *mymodel.MqttConnectOption) []func() {
	subers := make([]func(), 0)
	for _, topic := range mqttConf.Topics {
		subers = append(subers, func() {
			AeMqttClientSubscribe(topic, 0, f1, mqttConf, func(err error) {
				fmt.Println("subscribe1 error：")
			})
		})
	}
	return subers
}

// 这个是mqtt能接收也能发送对象的初始
func (o *AeEmqxObj) NewAeMqttClient(ctx context.Context, mqttConf *mymodel.MqttConnectOption) {
	mqttconfKey := mqttConf.Host
	aeMqttClient, exists := aeMqttClients[mqttconfKey]
	if exists {
		// 如果存在，不需要再重新连接了
		return
	}
	if !exists {
		//mqtt.DEBUG = log.New(os.Stdout, "", 0)
		mqtt.ERROR = log.New(os.Stdout, "", 0)
		opts := mqtt.NewClientOptions()
		broker := "tcp://" + mqttConf.Host + ":" + strconv.Itoa(mqttConf.Port)
		opts.AddBroker(broker)
		mqttClientId := mqttConf.ClientId
		if mqttClientId == "" {
			mqttClientId = appUtil.GenerateMqttClientId("rw")
			mqttConf.ClientId = mqttClientId
		}
		opts.SetClientID(mqttClientId)
		opts.SetUsername(mqttConf.User)
		opts.SetPassword(mqttConf.Password)
		opts.SetKeepAlive(60 * time.Second)
		// 设置消息回调处理函数
		opts.SetDefaultPublishHandler(f1)
		opts.SetPingTimeout(5 * time.Second)

		opts.OnConnect = func(c mqtt.Client) {
			glog.Line().Info(ctx, "MQTT connect ok")
			// 订阅事件
			glog.Line().Info(ctx, "MQTT subscribes！")
			subscribes := getSubscribers(ctx, mqttConf)
			for _, subscribe := range subscribes {
				subscribe()
			}
		}
		opts.OnConnectionLost = func(c mqtt.Client, err error) {
			glog.Info(ctx, "MQTT reconnect")
			time.Sleep(5 * time.Second)
			//aeMqttClient = nil
			o.NewAeMqttClient(ctx, mqttConf)
		}
		aeMqttClient = mqtt.NewClient(opts)
		aeMqttClients[mqttconfKey] = aeMqttClient
	}

	if token := aeMqttClient.Connect(); token.Wait() && token.Error() != nil {
		glog.Error(ctx, token.Error())
	}
}

// 这个是mqtt只发送，不订阅
func (o *AeEmqxObj) NewAeMqttOnlySendClient(ctx context.Context, mqttConf *mymodel.MqttConnectOption) {
	mqttconfKey := "send" + mqttConf.Host
	aeMqttClient, exists := aeMqttClients[mqttconfKey]
	if exists {
		// 如果存在，不需要再重新连接了
		return
	}
	if !exists {
		//mqtt.DEBUG = log.New(os.Stdout, "", 0)
		mqtt.ERROR = log.New(os.Stdout, "", 0)
		opts := mqtt.NewClientOptions()

		broker := "tcp://" + mqttConf.Host + ":" + strconv.Itoa(mqttConf.Port)
		opts.AddBroker(broker)

		mqttClientId := mqttConf.ClientId
		if mqttClientId == "" {
			mqttClientId = appUtil.GenerateMqttClientId("w")
			mqttConf.ClientId = mqttClientId
		}
		opts.SetClientID(mqttClientId)
		opts.SetUsername(mqttConf.User)
		opts.SetPassword(mqttConf.Password)
		opts.SetKeepAlive(60 * time.Second)
		// 设置消息回调处理函数
		opts.SetDefaultPublishHandler(f1)
		opts.SetPingTimeout(5 * time.Second)

		opts.OnConnect = func(c mqtt.Client) {
			glog.Line().Info(ctx, "MQTT connect ok and not subscribe ")
		}
		opts.OnConnectionLost = func(c mqtt.Client, err error) {
			glog.Info(ctx, "MQTT reconnect")
			time.Sleep(5 * time.Second)
			//aeMqttClient = nil
			o.NewAeMqttOnlySendClient(ctx, mqttConf)
		}
		aeMqttClient = mqtt.NewClient(opts)
		aeMqttClients[mqttconfKey] = aeMqttClient
	}

	if token := aeMqttClient.Connect(); token.Wait() && token.Error() != nil {
		glog.Error(ctx, token.Error())
	}
}

// 断开连接，主要是断掉订阅的实例，只发送的实例就不用管
func (o *AeEmqxObj) Disconnect(ctx context.Context, mqttConf *mymodel.MqttConnectOption) {
	mqttconfKey := mqttConf.Host
	aeMqttClient, exists := aeMqttClients[mqttconfKey]
	if exists {
		aeMqttClient.Disconnect(100) //100ms后才关闭
		go func() {
			// 实例清掉
			time.Sleep(100 * time.Duration(time.Millisecond))
			aeMqttClients[mqttconfKey] = nil
		}()
	}
}

// ClientSend 发布消息 ClientSend("topic/publish/example", 2, false, `{"code":0, "msg":"hello world!"}`)
func (o *AeEmqxObj) ClientSend(ctx context.Context, mqttConf *mymodel.MqttConnectOption, topic string, qos byte, retained bool, payload interface{}) error {
	mqttconfKey := "send" + mqttConf.Host
	aeMqttSendClient, exists := aeMqttClients[mqttconfKey]
	if !exists {
		o.NewAeMqttOnlySendClient(ctx, mqttConf)
		aeMqttSendClient, exists = aeMqttClients[mqttconfKey]
		time.Sleep(200 * time.Duration(time.Millisecond))
	}
	if !exists {
		glog.Error(ctx, "mqtt cannot send")
		return errors.New("mqtt cannot send")
	}
	if token := aeMqttSendClient.Publish(topic, qos, retained, payload); token.Wait() && token.Error() != nil {
		glog.Error(ctx, "send mqtt err", token.Error())
		return token.Error()
	}
	return nil
}

// 订阅消息
func AeMqttClientSubscribe(topic string, qos byte, callback mqtt.MessageHandler, mqttConf *mymodel.MqttConnectOption, err func(error)) {
	mqttconfKey := mqttConf.Host
	aeMqttClient, exists := aeMqttClients[mqttconfKey]
	if exists {
		if token := aeMqttClient.Subscribe(topic, qos, func(c mqtt.Client, msg mqtt.Message) {
			callback(c, msg)
		}); token.Wait() && token.Error() != nil {
			err(token.Error())
		}
	}

}
